package com.service;

import com.client.RepChainClient;
import com.dao.BaseDao;
import com.dao.SyncDao;
import com.protos.Peer;
import com.utils.LogUtil;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URI;
import java.nio.ByteBuffer;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Date;
import java.util.Timer;
import java.util.TimerTask;

/**
 * 增强同步服务
 * @author zyf
 */
public class SynServiceEnhance extends LogUtil {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    private String host;
    private static WebSocketClient client = null;
    private String propsPath;
//    private BaseDao baseDao;
    private Connection conn;
    private SyncDao syncDao;
    private Boolean isPull = false;
    private Boolean isSub = false;

    public SynServiceEnhance(String host, String propsPath) throws SQLException {
        this.host = host;
        this.propsPath = propsPath;
        this.conn = new BaseDao(propsPath).getConnectionFromDruid();
        this.syncDao = new SyncDao(conn);
        this.syncDao.InitSyncDao(host);
        this.syncDao.setUseTransaction(Boolean.TRUE);
    }

    /**
     * 拉取服务
     */
    public void pullBlocks() throws SQLException{
        this.isPull = true;
        Integer localHeight = this.syncDao.getLastHeight();
        logMsg(INFO,"本地已同步高度为：{}", localHeight);
        RepChainClient client = new RepChainClient(this.host);
        Integer remoteHeight = client.getChainHeight();
        logMsg(INFO,"远端链的高度为：{}", remoteHeight);
        if (remoteHeight != null) {
            while (localHeight < remoteHeight) {
                logMsg(INFO, "准备拉取块：{}", (localHeight + 1));
                this.syncDao.setBlockCount(remoteHeight);
                Peer.Block block = client.getBlockByHeight(localHeight + 1);
                try {
                    Integer rs = this.syncDao.saveBlock(block);
                    if (rs == 1) {
                        localHeight++;
                        logMsg(INFO, "更新后本地高度为：{}", localHeight);
                    } /*else {
                        this.isPull = false;
                    }*/
                } catch (SQLException e) {
                    if (this.syncDao.getUseTransaction()) {
                        this.syncDao.getConn().rollback();
                        logMsg(ERROR,"事务回滚：回滚block、transaction、synclocal表的操作");
                    }
                    e.printStackTrace();
                }
            }
            this.isPull = false;
            logMsg(INFO,"----------同步拉取结束----------");
        }
    }

    /**
     * 订阅服务
     */
    public WebSocketClient subBlocks() {
        final SynServiceEnhance synServEnhance = this;
        WebSocketClient client = new WebSocketClient(URI.create("ws://" + host + "/event")) {
            @Override
            public void onOpen(ServerHandshake serverHandshake) {
                logMsg(INFO, "webSocket连接已建立");
            }

            @Override
            public void onMessage(ByteBuffer message) {
                try {
                    byte[] bytes = new byte[message.capacity()];
                    message.get(bytes, 0, bytes.length);
                    Peer.Event event = Peer.Event.parseFrom(bytes);
                    if (event.hasBlk()) {
                        logMsg(INFO, "检测到块事件，准备启动拉取服务");
                        synServEnhance.syncDao.blockCountAdd();
                        if (!synServEnhance.isPull) {
                            synServEnhance.isSub = Boolean.FALSE;
                            logMsg(INFO,"----------开始拉取服务----------");
                            synServEnhance.pullBlocks();
                        } else {
                            synServEnhance.isSub = Boolean.TRUE;
                        }
                    }
                } catch (Exception ex) {
                    logMsg(ERROR, ex.getMessage());
                }
            }

            @Override
            public void onMessage(String s) {

            }

            @Override
            public void onClose(int i, String s, boolean b) {

            }

            @Override
            public void onError(Exception e) {
                logMsg(ERROR, "ws抛错: {}", e.getMessage());
            }
        };
        client.connect();
        return client;
    }

    // 避免线程被阻塞，定时启动两个线程？
    public void startSync() {
        Timer timer = new Timer("SynServiceEnhance--Iimer");
        final SynServiceEnhance synServiceEnhance = this;
        TimerTask pullTask = new TimerTask() {
            @Override
            public void run() {
                // scheduledExecutionTime() 返回此任务最近开始执行的时间
                Date date = new Date(this.scheduledExecutionTime());
                logMsg(INFO, "pullTask run {}", date);
                logMsg(INFO, "----------开始拉取服务----------");
                try {
                    synServiceEnhance.pullBlocks();
                    synServiceEnhance.isPull = Boolean.FALSE;
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
        TimerTask subTask = new TimerTask() {
            @Override
            public void run() {
                // scheduledExecutionTime() 返回此任务最近开始执行的时间
                Date date = new Date(this.scheduledExecutionTime());
                logMsg(INFO, "subTask run {}", date);
                logMsg(INFO, "开始启动监测Event");
                SynServiceEnhance.client = synServiceEnhance.subBlocks();
            }
        };
        TimerTask heart = new TimerTask() {
            @Override
            public void run() {
                Date date = new Date(this.scheduledExecutionTime());
                logMsg(INFO, "heart run {}", date);
                WebSocketClient client = SynServiceEnhance.client;
                logMsg(TRACE, client.toString());
                if (client.isClosed() | client.getConnection().isClosed() | client.getSocket().isClosed()) {
                    logMsg(INFO, "ws连接已经断掉");
                    try {
                        client.reconnect();
                    } finally {
                        client.reconnect();
                    }
                }
            }
        };
        timer.schedule(pullTask, 2000);
        timer.schedule(subTask, 8000);
        // 心跳监测socket连接是否断掉,1分钟监测一次
        timer.schedule(heart, 10000,60*1000);
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public String getPropsPath() {
        return propsPath;
    }

    public void setPropsPath(String propsPath) {
        this.propsPath = propsPath;
    }

}
